package com.lry.producer;

import com.lry.BrokerInfo;
import com.lry.RegisterType;
import com.lry.client.RpcClient;
import com.lry.client.RpcProxy;
import com.lry.message.Message;
import com.lry.message.MessageQueue;
import com.lry.raft.MqRaftService;
import com.lry.router.BrokerInfoChangeListener;
import com.lry.router.RandomRouter;
import com.lry.router.Subscribe;
import com.lry.rpc.MqRpcService;
import com.lry.selector.MessageSelector;
import com.lry.topic.TopicPublishInfo;
import lombok.extern.slf4j.Slf4j;
import net.sf.cglib.core.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.Collection;
import java.util.List;

@Slf4j
public class DefaultMqProducer implements MqProducer, BrokerInfoChangeListener {

    private String nameSrv;

    //所有主broker
    private List<RpcClient> rpcClients = new ArrayList<>();

    private boolean usingRaft;

    private String registerType;

    private Subscribe subscribe;

    private RandomRouter router = new RandomRouter();

    //单机版构造器，namesrv直接就是broker的地址
    public DefaultMqProducer(String nameSrv){
        this(nameSrv,false);
    }


    public DefaultMqProducer(String nameSrv,boolean usingRaft){
       this(nameSrv,RegisterType.ZOOKEEPER,usingRaft);
    }

    //如果使用了raft，则namesrv是注册中心的地址
    public DefaultMqProducer(String nameSrv,String registerType,boolean usingRaft){
        assert StringUtils.isNotEmpty(nameSrv);
        assert StringUtils.isNotEmpty(registerType);
        this.nameSrv = nameSrv;
        this.registerType = registerType;
        this.usingRaft = usingRaft;
    }

    @Override
    public void start() {
        if(usingRaft){
            this.subscribe = new Subscribe(registerType,nameSrv,this);
            subscribe.subscribe();

            //和所有主节点建立连接
            List<BrokerInfo> masters = subscribe.getMasters();
            for (BrokerInfo master : masters) {
                RpcClient rpcClient = new RpcClient(master.getHost(),master.getPort());
                rpcClients.add(rpcClient);
            }
        }else{
            String[] split = this.nameSrv.split(":");
            RpcClient rpcClient = new RpcClient(split[0],Integer.valueOf(split[1]));
            rpcClients.add(rpcClient);
        }
    }

    @Override
    public void shutdown() {

    }

    @Override
    public SendResult send(Message message) {
        RpcClient select = router.select(rpcClients);

        if(usingRaft){
            MqRaftService mqRaftService = RpcProxy.getProxy(select, MqRaftService.class);
            return mqRaftService.sendMsg(message);
        }else{
            MqRpcService mqRpcService = RpcProxy.getProxy(select, MqRpcService.class);
            return mqRpcService.sendMsg(message);
        }
    }

    @Override
    public SendResult send(Message message, MessageSelector messageSelector, Object arg) {

        MqRpcService mqRpcService = RpcProxy.getProxy(rpcClients.get(0), MqRpcService.class);

        TopicPublishInfo topicPublishInfo = mqRpcService.tryToFindTopicPublishInfo(message.getTopic());

        if(null==topicPublishInfo||null==topicPublishInfo.getMessageQueueList()){
            return SendResult.builder().sendStatus(SendStatus.SEND_FAIL).build();
        }

        List<MessageQueue> messageQueueList = topicPublishInfo.getMessageQueueList();
        MessageQueue select = messageSelector.select(messageQueueList, message, arg);

        return mqRpcService.sendMsg(message,select.getQueueId());
    }


    @Override
    public void change(BrokerInfo brokerInfo) {

    }
}
